import logging
from typing import Any, Dict, List, Literal, Optional, Type, Union

from pydantic import BaseModel
from sqlalchemy import BooleanClauseList, ColumnElement, Select
from sqlmodel import SQLModel, and_, inspect

logger = logging.getLogger(__name__)

# 条件类型：支持字典或 SQLAlchemy 表达式
ConditionType = Union[Dict[str, Any], BooleanClauseList, ColumnElement[bool]]

# 排序方向类型
SortOrderType = Literal["asc", "desc"]


def get_pk_names(model: Type[SQLModel]) -> List[str]:
    """
    获取模型的主键字段名称列表。
    Args:
        model: 模型类

    Returns:
        主键字段名称列表

    Raises:
        ValueError: 当模型没有定义主键时
    """
    insp = inspect(model)
    if not insp or not insp.primary_key:
        raise ValueError(f"Model {model.__name__} has no primary key defined")
    return [col.name for col in insp.primary_key]


def build_conditions(statement: Select, model: Type[SQLModel], conditions: Optional[ConditionType] = None) -> Select:
    """
    构建查询条件。

    Args:
        statement: SQLAlchemy 查询语句
        model: 模型类
        conditions: 查询条件，支持字典或SQLAlchemy表达式

    Returns:
        添加了条件的查询语句
    """
    if not conditions:
        return statement

    try:
        if isinstance(conditions, dict):
            # 处理字典条件
            where_clauses = []
            for key, value in conditions.items():
                if not hasattr(model, key):
                    logger.warning(f"Ignoring non-existent field in condition: {key}")
                    continue
                if value is None:
                    continue

                column = getattr(model, key)
                if isinstance(value, (list, tuple)) and value:
                    # IN 查询
                    where_clauses.append(column.in_(value))
                elif isinstance(value, dict):
                    # 支持操作符: {'gt': 5, 'lt': 10}
                    for op, op_value in value.items():
                        if op == "gt":
                            where_clauses.append(column > op_value)
                        elif op == "gte":
                            where_clauses.append(column >= op_value)
                        elif op == "lt":
                            where_clauses.append(column < op_value)
                        elif op == "lte":
                            where_clauses.append(column <= op_value)
                        elif op == "ne":
                            where_clauses.append(column != op_value)
                        elif op == "like":
                            where_clauses.append(column.like(f"%{op_value}%"))
                        elif op == "ilike":
                            where_clauses.append(column.ilike(f"%{op_value}%"))
                else:
                    # 等值查询
                    where_clauses.append(column == value)

            if where_clauses:
                statement = statement.where(and_(*where_clauses))
        else:
            # 直接使用SQLAlchemy表达式
            statement = statement.where(conditions)

    except Exception as e:
        logger.error(f"Error building conditions: {e}")
        raise

    return statement


def apply_soft_delete_filter(
    statement: Select, model: Type[SQLModel], include_deleted: bool = False, deleted_field: str = "is_deleted"
) -> Select:
    """
    应用软删除过滤条件。

    Args:
        statement: 查询语句
        model: 模型类
        include_deleted: 是否包含已删除的记录
        deleted_field: 软删除字段名

    Returns:
        添加了软删除过滤条件的查询语句
    """
    if not include_deleted and hasattr(model, deleted_field):
        statement = statement.where(getattr(model, deleted_field) == 0)
    return statement


def apply_sorting(
    statement: Select,
    model: Type[SQLModel],
    sort_by: Optional[List[str]] = None,
    sort_order: Optional[List[SortOrderType]] = None,
) -> Select:
    """
    应用排序条件到查询语句。

    Args:
        statement: 查询语句
        model: 模型类
        sort_by: 排序字段列表
        sort_order: 排序方向列表

    Returns:
        添加了排序的查询语句

    Raises:
        ValueError: 当排序字段不存在于模型中时
    """
    if not sort_by:
        return statement

    # 验证排序字段
    invalid_fields = [field for field in sort_by if not hasattr(model, field)]
    if invalid_fields:
        raise ValueError(f"Sorting fields do not exist in model: {invalid_fields}")

    # 处理排序方向
    if not sort_order:
        sort_order = ["asc"] * len(sort_by)
    elif len(sort_by) != len(sort_order):
        logger.warning("Length of sort_by and sort_order mismatch; using 'asc' for all.")
        sort_order = ["asc"] * len(sort_by)

    # 构建排序表达式
    sort_expressions = []
    for field, direction in zip(sort_by, sort_order):
        column = getattr(model, field)
        if direction.lower() == "desc":
            sort_expressions.append(column.desc())
        else:
            sort_expressions.append(column.asc())

    return statement.order_by(*sort_expressions)


def validate_pk_fields(obj: BaseModel, model: Type[SQLModel]) -> None:
    """
    验证对象是否包含所有主键字段。

    Args:
        obj: 包含主键字段的对象
        model: 模型类

    Raises:
        ValueError: 当缺少主键字段时
    """
    _pk_names = get_pk_names(model)
    if not _pk_names:
        raise ValueError(f"Model {model.__name__} has no primary key defined")

    missing_pks = [pk for pk in _pk_names if not hasattr(obj, pk)]
    if missing_pks:
        raise ValueError(f"Missing primary key fields {missing_pks} in {obj.__class__.__name__}")


def extract_pk_conditions(obj: BaseModel, model: Type[SQLModel]) -> BooleanClauseList:
    """
    从对象中提取主键条件。

    Args:
        obj: 包含主键值的对象
        model: 模型类

    Returns:
        主键条件的SQLAlchemy表达式
    """
    validate_pk_fields(obj, model)
    return and_(getattr(model, pk) == getattr(obj, pk) for pk in get_pk_names(model))
